package com.grf.kafka.java.simple;

import ch.qos.logback.classic.Logger;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

/**
 * @apiNote 生产者
 * @author grf
 * @since 2024/3/29
 */
@Slf4j
public class Producer {
    public static final String BOOTSTRAP_SERVERS = "127.0.0.1:9092";

    public static final String topic = "disTopic";
    @SneakyThrows
    public static void main(String[] args) {
        //PART1:设置发送者相关属性
        Properties props = new Properties();
        // 此处配置的kafka的端口
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        // 配置key的序列化类
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        org.apache.kafka.clients.producer.Producer<String, String> producer = new KafkaProducer<>(props);
        CountDownLatch latch = new CountDownLatch(5);
        for (int i=0; i<10; i++){
            // part2: 构建消息
            ProducerRecord<String, String> record = new ProducerRecord<>(topic,Integer.toString(i), "MyProducer "+i);
            // part3: 发送消息
            // 单向发送，不关心服务应答
            // producer.send(record);
            // log.info("message {} send",i);
            // 同步发送，获取服务端应答消息前，会阻塞当前线程
            RecordMetadata metadata = producer.send(record).get();
            String topic = metadata.topic();
            int partition = metadata.partition();
            long offset = metadata.offset();
            String message = metadata.toString();
            log.info("topic is {}; partition is {}; offset is {}; message is {}", topic, partition, offset, message);
            // // 异步发送，消息发送后不阻塞，服务端有应答后触发回调函数
            // producer.send(record, new Callback() {
            //     @Override
            //     public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            //         if (e != null) {
            //             System.out.println("消息发送失败，");
            //             e.printStackTrace();
            //         } else {
            //             String topic = recordMetadata.topic();
            //             long offset = recordMetadata.offset();
            //             String message = recordMetadata.toString();
            //             System.out.println("message : ["+message+"] sended with topic:" + topic+"; offset:"+offset);
            //         }
            //     }
            // });
            latch.countDown();
        }
        // 消息处理完停止发送
        latch.await();
        producer.close();
    }
    /**
     * 禁用debug日志
     */
    static {
        ch.qos.logback.classic.LoggerContext loggerContext = (ch.qos.logback.classic.LoggerContext) org.slf4j.LoggerFactory.getILoggerFactory();
        List<Logger> loggerList = loggerContext.getLoggerList();
        loggerList.forEach(logger -> logger.setLevel(ch.qos.logback.classic.Level.INFO));
    }
}
